package com.zhs;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

@Getter
@Setter
@Slf4j
public class SpringConsumer {
    private String consumerGroupName;
    private String nameServer;
    private String topicName;
    private DefaultMQPushConsumer consumer;
    private MessageListenerConcurrently messageListenerConcurrently;
    public SpringConsumer(String consumerGroupName,String nameServer,String topicName,MessageListenerConcurrently messageListenerConcurrently){
        this.consumerGroupName=consumerGroupName;
        this.topicName=topicName;
        this.messageListenerConcurrently=messageListenerConcurrently;
    }
    public void init() throws Exception{
        log.info("开始启动消息消费者！");
        consumer=new DefaultMQPushConsumer(consumerGroupName);
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(topicName,"*");
        consumer.registerMessageListener(messageListenerConcurrently);
        consumer.start();
        log.info("消息消费者启动成功！");
    }
    public void destory(){
        log.info("开始关闭消息消费者！");
        consumer.shutdown();
        log.info("成功关闭消息消费者！");
    }
}
